package drds.binlog.metadata;

import com.google.common.collect.Lists;
import com.google.common.collect.MapMaker;
import com.google.common.collect.Maps;
import drds.binlog.common.Authors;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.position.PositionRange;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

@Author(name = Authors.LI_YANG)
public class BatchIdToPositionRangeConcurrentMap
{

    @Setter
    @Getter
    private ClientInfo clientInfo;
    @Setter
    @Getter
    private Map<Long, PositionRange> batchIdToPositionRangeMap = new MapMaker().makeMap();
    private AtomicLong batchId = new AtomicLong(1);

    public BatchIdToPositionRangeConcurrentMap()
    {

    }

    protected BatchIdToPositionRangeConcurrentMap(ClientInfo clientInfo)
    {
        this.clientInfo = clientInfo;
    }

    public static BatchIdToPositionRangeConcurrentMap create(ClientInfo clientInfo)
    {
        return new BatchIdToPositionRangeConcurrentMap(clientInfo);
    }

    public synchronized void addPositionRange(Long batchId, PositionRange positionRange)
    {
        updateBatchId(batchId);
        batchIdToPositionRangeMap.put(batchId, positionRange);
    }

    public synchronized Long addPositionRange(PositionRange positionRange)
    {
        Long batchId = this.batchId.getAndIncrement();
        batchIdToPositionRangeMap.put(batchId, positionRange);
        return batchId;
    }

    public synchronized PositionRange removePositionRange(Long batchId)
    {
        if (batchIdToPositionRangeMap.containsKey(batchId))
        {
            Long minBatchId = Collections.min(batchIdToPositionRangeMap.keySet());
            if (!minBatchId.equals(batchId))
            {
                // 检查一下提交的ack/rollback，必须按batchId分出去的顺序提交，否则容易出现丢数据
                throw new MetaDataManagerException(String.format("batchId:%d is not the firstly:%d",
                        batchId,
                        minBatchId));
            }
            return batchIdToPositionRangeMap.remove(batchId);
        } else
        {
            return null;
        }
    }

    public synchronized PositionRange getPositionRange(Long batchId)
    {
        return batchIdToPositionRangeMap.get(batchId);
    }


    public synchronized PositionRange getFirstPositionRange()
    {
        if (batchIdToPositionRangeMap.size() == 0)
        {
            return null;
        } else
        {
            Long batchId = Collections.min(batchIdToPositionRangeMap.keySet());
            return batchIdToPositionRangeMap.get(batchId);
        }
    }

    public synchronized PositionRange getLastestPositionRange()
    {
        if (batchIdToPositionRangeMap.size() == 0)
        {
            return null;
        } else
        {
            Long batchId = Collections.max(batchIdToPositionRangeMap.keySet());
            return batchIdToPositionRangeMap.get(batchId);
        }
    }

    public synchronized Map<Long, PositionRange> getBatchIdToPositionRangeMap()
    {
        Set<Long> batchIdSets = batchIdToPositionRangeMap.keySet();
        List<Long> batchIdList = Lists.newArrayList(batchIdSets);
        Collections.sort(Lists.newArrayList(batchIdList));

        return Maps.newHashMap(batchIdToPositionRangeMap);
    }

    public synchronized void clear()
    {
        batchIdToPositionRangeMap.clear();
    }

    private synchronized void updateBatchId(Long batchId)
    {
        if (this.batchId.get() < batchId + 1)
        {
            this.batchId.set(batchId + 1);
        }
    }


}
